package main

import (
	"fmt"
	"log"

	"github.com/louloulin/dataflare/pkg/workflow/processor"
)

func main() {
	fmt.Println("=== DataFlare Processors Verification ===")
	
	// Verify transform processor
	verifyTransformProcessor()
	
	// Verify filter processor
	verifyFilterProcessor()
	
	fmt.Println("\n=== Processors Verification Complete ===")
}

// Verify transform processor
func verifyTransformProcessor() {
	fmt.Println("\n=== Verifying Transform Processor ===")
	
	// Create transform processor
	fmt.Println("Creating transform processor...")
	transformProcessor, err := processor.NewProcessor("transform")
	if err != nil {
		log.Fatalf("Failed to create transform processor: %v", err)
	}
	
	// Initialize processor with transform script
	err = transformProcessor.Init(map[string]interface{}{
		"function": `
			function process(data) {
				// Parse the data
				const obj = JSON.parse(data);
				
				// Double the value
				obj.value = obj.value * 2;
				
				// Add processed flag
				obj.processed = true;
				
				// Return the processed data
				return JSON.stringify(obj);
			}
		`,
	})
	if err != nil {
		log.Fatalf("Failed to initialize transform processor: %v", err)
	}
	
	// Test data
	testData := []string{
		`{"id": 1, "value": 10}`,
		`{"id": 2, "value": 20}`,
		`{"id": 3, "value": 30}`,
	}
	
	// Process data
	fmt.Println("\nProcessing data:")
	for i, data := range testData {
		// Process data
		processedData, err := transformProcessor.Process([]byte(data), nil)
		if err != nil {
			log.Fatalf("Failed to process data: %v", err)
		}
		
		fmt.Printf("Record %d:\n", i+1)
		fmt.Printf("  Original: %s\n", data)
		fmt.Printf("  Processed: %s\n", string(processedData))
	}
	
	// Close processor
	err = transformProcessor.Close()
	if err != nil {
		log.Fatalf("Failed to close processor: %v", err)
	}
	
	fmt.Println("Transform processor verification completed successfully!")
}

// Verify filter processor
func verifyFilterProcessor() {
	fmt.Println("\n=== Verifying Filter Processor ===")
	
	// Create filter processor
	fmt.Println("Creating filter processor...")
	filterProcessor, err := processor.NewProcessor("filter")
	if err != nil {
		log.Fatalf("Failed to create filter processor: %v", err)
	}
	
	// Initialize processor with filter script
	err = filterProcessor.Init(map[string]interface{}{
		"function": `
			function process(data) {
				// Parse the data
				const obj = JSON.parse(data);
				
				// Filter condition: value > 15
				if (obj.value > 15) {
					return data;
				} else {
					return null;
				}
			}
		`,
	})
	if err != nil {
		log.Fatalf("Failed to initialize filter processor: %v", err)
	}
	
	// Test data
	testData := []string{
		`{"id": 1, "value": 10}`,
		`{"id": 2, "value": 20}`,
		`{"id": 3, "value": 30}`,
	}
	
	// Process data
	fmt.Println("\nProcessing data:")
	for i, data := range testData {
		// Process data
		processedData, err := filterProcessor.Process([]byte(data), nil)
		if err != nil {
			log.Fatalf("Failed to process data: %v", err)
		}
		
		fmt.Printf("Record %d:\n", i+1)
		fmt.Printf("  Original: %s\n", data)
		
		if processedData == nil {
			fmt.Printf("  Filtered out\n")
		} else {
			fmt.Printf("  Passed filter: %s\n", string(processedData))
		}
	}
	
	// Close processor
	err = filterProcessor.Close()
	if err != nil {
		log.Fatalf("Failed to close processor: %v", err)
	}
	
	fmt.Println("Filter processor verification completed successfully!")
}


select ord_date,count(*) from datahub.hedge_trade_record where ord_date >='20250328' 